/* file name  : Dispatch.java
 * created    : Fri 23 Sep 2005 10:58:53 PM EDT
 * authors    : Saikat Guha <saikat@cs.cornell.edu>
 * copyright  : GNU Lesser General Public License version 2.1 or later
 * 
 * Copyright (C) 2005 Saikat Guha <saikat@cs.cornell.edu>
 * 
 * This library is free software; you can redistribute it and/or modify it under the terms
 * of the GNU Lesser General Public License as published by the Free Software Foundation;
 * either version 2.1 of the License, or (at your option) any later version.
 * 
 * This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
 *  without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * See the GNU Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public License along with this
 * library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330,
 * Boston, MA 02111-1307 USA
 * 
 */
package net.nutss;

import java.util.*;

import net.nutss.stunt.STUNTServerSocket;
import net.nutss.stunt.STUNTSocket;
import net.nutss.stunt.STUNTUDPSocket;

import static net.nutss.Control.log;

/**
 * Dispatches message events to running handlers
 * 
 * @author Saikat Guha
 * @version %I%, %G%
 */
public abstract class Dispatch extends Thread {

	/**
	 * Signaling channel
	 */
	protected final SignalingContext sig;
	public STUNTSelector sel=new STUNTSelector();

	/**
	 * Kkey for signaling channel events
	 */

	/**
	 * Continuations indexed by signaling messages that they subscribe to
	 */
	private HashMap<SignalingMessage, Continuation> continuations = new HashMap<SignalingMessage, Continuation>();

	/**
	 * Continuations in order of when their timeouts expire
	 */
	private TreeMap<Long, LinkedList<Continuation>> timers = new TreeMap<Long, LinkedList<Continuation>>();

	/**
	 * Reverse map from continuations to their timeout expiry
	 */
	private HashMap<Continuation, LinkedList<Continuation>> revTimers = new HashMap<Continuation, LinkedList<Continuation>>();

	/**
	 * Reverse map from continuations to the signaling messages that they
	 * subscribe to
	 */
	private HashMap<Continuation, SignalingMessage> revContinuations = new HashMap<Continuation, SignalingMessage>();

	/**
	 * System time
	 */
	private long clock;

	/**
	 * Selector for non--blocking IO
	 */
	private Object o1 = new Object();
	private boolean isTerminstate = true;

	public void setTerm(boolean val) {
		synchronized (o1) {
			isTerminstate = val;
		}
	}

	public boolean getTerm() {
		boolean bval;
		synchronized (o1) {
			bval = isTerminstate;
		}
		return bval;
	}

	/**
	 * Registers a timeout for a continuation
	 * 
	 * @param tm
	 *            time at which timeout expires in milliseconds since epoch
	 * @param k
	 *            the continuation
	 */
	private void addTimer(Long tm, Continuation k) {
		LinkedList<Continuation> l = timers.get(tm);
		if (l == null) {
			l = new LinkedList<Continuation>();
			timers.put(tm, l);
		}
		l.add(k);
		revTimers.put(k, l);
	}

	/**
	 * Adds a new continuation to the dispatcher
	 * 
	 * @param k
	 *            the continuation
	 * @param p
	 *            the message creating the continuation, null of none
	 */
	protected void addContinuation(Continuation k, SignalingMessage p) {
		SignalingMessage interest = k.init(p);
		continuations.put(interest, k);
		revContinuations.put(k, interest);
		revTimers.remove(k);
		addTimer(0L, k);
	}

	/**
	 * Cancels a continuation
	 * 
	 * @param k
	 *            the continuation
	 */
	private void cancel(Continuation k) {
		k.cancel(this);
		continuations.remove(revContinuations.get(k));
		revContinuations.remove(k);
	}

	/**
	 * Creates a new dispatcher for a given signaling channel
	 */
	protected Dispatch(SignalingContext context) {
		try {
//			this.sigKey = context.registerChannel(selector);
		} catch (Exception e) {
			log.throwing("Dispatch", "Dispatch", e);
		} finally {
			this.sig = context;
		}
	}

	/**
	 * Returns the non-blocking socket event selector
	 * 
	 * @return the selector
	 */

	/**
	 * Handles a signaling message not handled by any continuation
	 * 
	 * @param msg
	 */
	protected abstract void handleMessage(SignalingMessage msg);

	private void step(Continuation k, Object arg,STUNTSelector sel) {
		int tmp;
		//log.entering("Dispatch.step", k.toString(), arg);
		while ((tmp = k.step(arg, this,sel)) == 0)
			arg = null;
		//log.info("Dispatch.step"+tmp);
		if (tmp < 0)
			cancel(k);
		else
			addTimer(clock + tmp, k);
	}

	/**
	 * Allows the continuation to communicate with the dispatcher
	 * 
	 * @param k
	 *            the continuation
	 * @param o
	 *            the object being passed to the dispatcher
	 */
	public abstract void callback(Continuation k, Object o);

	public void run() {
		long tm;
		/*
		 * by casper not supported int num, tmp;
		 */
		setTerm(false);
		try {
			while (!getTerm() && !Thread.interrupted() && !timers.isEmpty()) {

				// Process timeouts in batches
				while (!timers.isEmpty()
						&& timers.firstKey() < (clock = System
								.currentTimeMillis() + 10)) {
					while (!timers.isEmpty()
							&& (tm = timers.firstKey()) < clock) {
						LinkedList<Continuation> l = timers.remove(tm);
						if (tm < 0)
							for (Continuation k : l)
								cancel(k);
						else
							for (Continuation k : l)
								step(k, null,sel);
					}
				}

				tm = timers.isEmpty() ? 1000 : timers.firstKey()
						- System.currentTimeMillis();
				ArrayList bList;
				long itm=tm;
				while ((bList=sel.getHasData()).size() == 0 && itm>0){
					try
					{
						Thread.sleep(itm>=25?25:itm);
						itm-=25;
					}
					catch(InterruptedException e)
					{
						
					}
				}
				//must check
				if(bList.size()==0) bList = sel.getHasData();
				if (bList.size() == 0)
					continue;
				if (bList.size() > 0) {
					for (int i = 0; i < bList.size(); ++i) {
						Continuation k = null;
						Object arg = null;
						STUNTSocket so = null;
						STUNTUDPSocket sudp=null;
						STUNTServerSocket ss = null;
						Object obj = bList.get(i);
						if (obj instanceof STUNTUDPSocket) {
							sudp = (STUNTUDPSocket) obj;
							k = sudp.con;
							// Activity on the signaling channel. It's some
							// ProxyMessage
							SignalingMessage msg = sig.receive(sudp.readBytes());
							log.finest(String.format("received msg=%s", msg));
							k = continuations.get(msg);
							if (k != null) {
								arg = msg;
							} else { // not claimed by existing continuations
								handleMessage(msg);
							}
						} else if (obj instanceof STUNTSocket) {
							so = (STUNTSocket) obj;
							k = so.con;
						} else if (obj instanceof STUNTServerSocket) {
							ss = (STUNTServerSocket) obj;
							k = ss.con;
						}
						if (k != null && sudp==null) {
							// Activity on an arbbbitrary key. Some direct
							// socket operation
							// finished. Make sure the continuation is still
							// valid.
							arg = obj;
							if (revContinuations.get(k) == null) {
								k = null;
								sel.remove(obj);
							}
						}
						if (k != null) {
							//log.info(k.toString());
							//log.info(arg.toString());
							revTimers.get(k).remove(k);
							step(k, arg,sel);
						}
					}
				}
				clock = System.currentTimeMillis();
				// Reap empty timer batches as continuations got pulled out in
				// the last pass
				while (!timers.isEmpty()
						&& timers.get(tm = timers.firstKey()).size() == 0)
					timers.remove(tm);
			}
			sig.close();
			setTerm(true);
		} catch (java.io.IOException e) {
			setTerm(true);
			log.throwing("Dispatch", "run", e);
		}

		// close any continuations waiting for inputs since the event handler is
		// terminating
		for (LinkedList<Continuation> l : timers.values()) {
			for (Continuation c : l) {
				c.cancel(this);
			}
		}
	}
}
